package net.xmmpp.uc.im.group.broker.consumer;

import com.alibaba.fastjson.JSONObject;
import com.google.common.hash.HashCode;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import net.xmmpp.uc.im.group.broker.bean.MessageXmpp;
import net.xmmpp.uc.im.common.constants.Constants;
import net.xmmpp.uc.im.common.model.ExtMsgLog;
import net.xmmpp.uc.im.common.utils.Utils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.dom4j.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.MimeTypeUtils;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;


@Service
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic}", consumerGroup = "${rocketmq.consumer.group}", consumeMode =  ConsumeMode.ORDERLY)
public class IMMucListener implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(IMMucListener.class);

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private RedisTemplate<String, Long> redisTemplate;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private static HashFunction hashFunction = Hashing.murmur3_128();

    @Value("${rocketmq.producer.topic}")
    private String outTopic;

    private final static String LOG_FROM = "FROM";
    private final static String LOG_TO = "TO";

    public void onMessage(MessageExt message) {
        String body = new String(message.getBody());
        MessageXmpp msgXmpp = JSONObject.parseObject(body, MessageXmpp.class);

        StringBuilder logBuilder = new StringBuilder();
        logBuilder.append("------------- BEGIN -------------").append('\n')
                .append("MsgId: " ).append(message.getMsgId()).append('\n')
                .append("From: " ).append(msgXmpp.getFrom()).append('\n')
                .append("To: " ).append(msgXmpp.getTo()).append('\n')
                .append("Ts: " ).append(msgXmpp.getTs()).append('\n')
                .append("XMPP: " ).append(msgXmpp.getXmpp()).append('\n');
//                .append("-------------  END  -------------");
        LOGGER.info(logBuilder.toString());

        Element rootElement = MsgDOMReader.getInstance().getRootElement(msgXmpp.getXmpp());
        Element childElement = MsgDOMReader.getInstance().getMsgFirstChild(rootElement);

        if(Constants.DOM_ELEMENT_BODY.equals(childElement.getName())){
            // 通用消息，所有用户都接收
            performMsg(msgXmpp.getFrom(), msgXmpp, rootElement, false, null, LOG_FROM);
            performMsg(msgXmpp.getTo(), msgXmpp, rootElement,false, null, LOG_TO);

        } else if(Constants.DOM_ELEMENT_SUBSCRIBE.equals(childElement.getName())){
            // 加好友接收者：收到如下消息; 加好友请求者：消息队列中不会有这条消息
            performMsg(msgXmpp.getTo(), msgXmpp, rootElement,false, null, LOG_FROM);

        } else if(Constants.DOM_ELEMENT_SUBSCRIBED.equals(childElement.getName())){
            String typeStr = MsgDOMReader.getInstance().getElementAttr(childElement, Constants.DOM_ATTR_TYPE);

            if(Constants.DOM_TYPE_RECIPIENT.equals(typeStr)){
                // 加好友接收者：收到如下消息：
                performMsg(msgXmpp.getFrom(), msgXmpp, rootElement, false, null,LOG_FROM);
            } else if(Constants.DOM_TYPE_SUBSCRIBE.equals(typeStr)){
                // 加好友请求者：收到如下消息：
                performMsg(msgXmpp.getTo(), msgXmpp, rootElement, true, Constants.ATTACHMENT_KEY_ROSTER_ADD_ITEM, LOG_TO);
            }
        } else if(Constants.DOM_ELEMENT_UNSUBSCRIBED.equals(childElement.getName())){
            // 加好友接收者：收到如下消息; 加好友请求者：消息队列中不会有这条消息
            performMsg(msgXmpp.getFrom(), msgXmpp, rootElement, false, null,LOG_FROM);

        } else if(Constants.DOM_ELEMENT_UNSUBSCRIBE.equals(childElement.getName())){
            String typeStr = MsgDOMReader.getInstance().getElementAttr(childElement, Constants.DOM_ATTR_TYPE);
            if(Constants.DOM_TYPE_UNSUBSCRIBE.equals(typeStr)){
                // 解除好友关系的接收者：收到如下消息; 发起者：消息队列中不会有这条消息
                performMsg(msgXmpp.getFrom(), msgXmpp, rootElement, false, null, LOG_FROM);
            } else if(Constants.DOM_TYPE_RECIPIENT.equals(typeStr)){
                // 解除好友关系的发起者：收到如下消息：
                performMsg(msgXmpp.getTo(), msgXmpp, rootElement, true, Constants.ATTACHMENT_KEY_ROSTER_UPDATE_ITEM, LOG_TO);
            }
        }
    }


    private Message<ExtMsgLog> performMsg(String uid, MessageXmpp msgXmpp, Element rootElement, boolean attached, String keyItem, String logFlag){
        String usrMsgVerKey = Utils.formatUsrMsgVerKey(uid);
        Long ver = redisTemplate.opsForValue().increment(usrMsgVerKey);
        String userMsgKey = Utils.formatUserMsgQueueId(uid);

        String xmpp =  MsgDOMReader.getInstance().setMsgVersion(rootElement, ver);
        stringRedisTemplate.opsForZSet().add(userMsgKey, xmpp, ver);

        ExtMsgLog msgLog = new ExtMsgLog();
        msgLog.setUid(uid);
        msgLog.setType(msgXmpp.getType());
        msgLog.setVer(ver);
        msgLog.setTs(msgXmpp.getTs());
        if(attached){
            Map<String, String> attachment = new HashMap<>();
            attachment.put(keyItem, xmpp);
            msgLog.setAttachment(attachment);
        }

        Message<ExtMsgLog> noticeMsg = MessageBuilder.withPayload(msgLog)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();

        StringBuilder logBuilder = new StringBuilder();
        logBuilder.append(logFlag)
                .append(" LOG - uid = ").append(uid).append(", ")
                .append(msgLog).append('\n');

        SendResult sendResult = null;
        try {
            sendResult = rocketMQTemplate.syncSendOrderly(chooseOutboundTopic(uid), noticeMsg, uid);
        } finally {
            logBuilder.append(logFlag)
                    .append(" LOG Result - ")
                    .append(sendResult);
            LOGGER.info(logBuilder.toString());
        }

        return noticeMsg;
    }


    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
//        String topic0 = "uc-in-topic-0";
//        String topic1 = "uc-in-topic-1";
//        try {
//            consumer.subscribe(topic0, "*");
//            consumer.subscribe(topic1, "*");
//        } catch (MQClientException e) {
//            LOGGER.error("Subscribe uc-in-topic ERROR", e);
//            throw new RuntimeException(e);
//        }

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    }

    private String chooseOutboundTopic(String uid){
        HashCode hashCode = hashFunction.hashString(uid, StandardCharsets.UTF_8);
        int index = hashCode.hashCode() % 2;
        String topic = outTopic + "-" + Math.abs(index);

        return topic;
    }


    /*
        String fromUsrMsgVerKey = Utils.formatUsrMsgVerKey(msgXmpp.getFrom());
        Long fromVer = redisTemplate.opsForValue().increment(fromUsrMsgVerKey);
        String fromUserMsgKey = Utils.formatUserMsgQueueId(msgXmpp.getFrom());

        String fromXmpp =  MsgDOMReader.getInstance().setMsgVersion(msgXmpp.getXmpp(), fromVer);
        stringRedisTemplate.opsForZSet().add(fromUserMsgKey, fromXmpp, fromVer);

        MsgLog fromLog = new MsgLog();
        fromLog.setUid(msgXmpp.getFrom());
        fromLog.setType(msgXmpp.getType());
        fromLog.setVer(fromVer);
        fromLog.setTs(msgXmpp.getTs());

        String toUsrMsgVerKey = Utils.formatUsrMsgVerKey(msgXmpp.getTo());
        Long toVer = redisTemplate.opsForValue().increment(toUsrMsgVerKey);
        String toUserMsgKey = Utils.formatUserMsgQueueId(msgXmpp.getTo());

        String toXmpp =  MsgDOMReader.getInstance().setMsgVersion(msgXmpp.getXmpp(), toVer);
        stringRedisTemplate.opsForZSet().add(toUserMsgKey, toXmpp, toVer);

        MsgLog toLog = new MsgLog();
        toLog.setUid(msgXmpp.getTo());
        toLog.setType(msgXmpp.getType());
        toLog.setVer(toVer);
        toLog.setTs(msgXmpp.getTs());

        Message<MsgLog> fromNoticeMsg = MessageBuilder.withPayload(fromLog)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();
        Message<MsgLog> toNoticeMsg = MessageBuilder.withPayload(toLog)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();

        LOGGER.info("fromLog - " + fromLog);
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(chooseOutboundTopic(fromLog.getUid()), fromNoticeMsg, fromLog.getUid());
        LOGGER.info("fromLog Result - " + sendResult);

        LOGGER.info("toLog - " + toLog);
        sendResult = rocketMQTemplate.syncSendOrderly(chooseOutboundTopic(toLog.getUid()), toNoticeMsg, toLog.getUid());
        LOGGER.info("toLog Result - " + sendResult);
    */
}


